package cn.doitedu.rtmk.mgmt.doit39_rtmk_management.controller;

import cn.doitedu.rtmk.mgmt.doit39_rtmk_management.pojo.Param;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONArray;
import com.alibaba.fastjson2.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.apache.http.HttpHost;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.index.query.*;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.joda.time.DateTimeUtils;
import org.roaringbitmap.longlong.Roaring64Bitmap;
import org.springframework.boot.configurationprocessor.json.JSONException;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import redis.clients.jedis.Jedis;

import java.io.ByteArrayOutputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.sql.*;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Date;

@Slf4j
@RestController
public class RuleManageController {

    Connection conn;
    RestHighLevelClient restClient;
    SearchRequest request;

    public RuleManageController() throws SQLException {
        conn = DriverManager.getConnection("jdbc:mysql://doitedu:3306/doit39", "root", "root");


        restClient = new RestHighLevelClient(
                RestClient.builder(new HttpHost("doitedu", 9200, "http"))
        );

        request = new SearchRequest("doit39_profile");

    }

    /**
     * 新增规则
     * @param param
     * @return
     * @throws JSONException
     * @throws IOException
     * @throws SQLException
     */
    @RequestMapping("/api/rule/add")
    public String addRule(@RequestBody Param param) throws JSONException, IOException, SQLException, ParseException {
        // 取出前端发过来的数据中的 规则参数Json
        String paramJson = param.getParamJson();
        String modelId = param.getModel_id();

        // 根据模型id，去模型meta表中查询该模型的运算机源代码
        PreparedStatement modelPst = conn.prepareStatement("select calculator_source_code from rule_model_meta where model_id = ?");
        modelPst.setString(1,modelId);
        // 执行查询
        ResultSet resultSet = modelPst.executeQuery();
        resultSet.next();
        String calculatorSourceCode = resultSet.getString("calculator_source_code");


        // 解析规则参数json
        JSONObject paramJsonObject = JSON.parseObject(paramJson);
        String ruleId = paramJsonObject.getString("rule_id");

        // 从中提取静态画像条件
        JSONArray offlineProfile = paramJsonObject.getJSONArray("offline_profile");


        BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
        // 遍历参数中的静态画像条件参数，构建es的查询条件
        for (int i = 0; i < offlineProfile.size(); i++) {
            JSONObject profile_i = offlineProfile.getJSONObject(i);

            String tagName = profile_i.getString("tag_name");
            String compareType = profile_i.getString("compare_type");
            JSONArray tagValueArray = profile_i.getJSONArray("tag_value");

            // 拼接这些条件到es的查询条件中
            QueryBuilder queryBuilder;
            if ("=".equals(compareType)) {
                queryBuilder = QueryBuilders.termQuery(tagName, tagValueArray.getString(0));
            } else if (">".equals(compareType)) {
                queryBuilder = QueryBuilders.rangeQuery(tagName).gt(tagValueArray.getDoubleValue(0));
            } else if ("match".equals(compareType)) {
                queryBuilder = QueryBuilders.matchQuery(tagName, tagValueArray.getString(0));
            } else {
                queryBuilder = QueryBuilders.termQuery(tagName, tagValueArray.getString(0));
            }

            // 将本次遍历到的参数所构建的query条件，放入总条件
            boolQueryBuilder.must(queryBuilder);
        }

        // 将拼装好的 es 查询条件，放入搜索请求对象
        SearchRequest request = this.request.source(new SearchSourceBuilder().query(boolQueryBuilder));

        // 用es客户端，执行该搜索请求
        SearchResponse response = restClient.search(request, RequestOptions.DEFAULT);
        SearchHits hits = response.getHits();


        Roaring64Bitmap targetUsersBitmap = Roaring64Bitmap.bitmapOf();
        // 迭代搜索结果中的每一个用户，并将用户id放入bitmap
        for (SearchHit hit : hits) {
            long targetUserId = Long.parseLong(hit.getId());
            System.out.println("------查询到一个目标用户: " + targetUserId);
            targetUsersBitmap.add(targetUserId);
        }

        // 将bitmap序列化
        ByteArrayOutputStream baout = new ByteArrayOutputStream();
        DataOutputStream dataOut = new DataOutputStream(baout);
        targetUsersBitmap.serialize(dataOut);
        byte[] targetUsersBitmapBytes = baout.toByteArray();


        // 为那些跨越上线时间点范围的条件，查询它的历史值
        /*
                     "online_history_profile":{
                              "event_id": "w",
                              "start_time":"2023-07-01 00:00:00",
                              "end_time":"9999-12-31 00:00:00",
                              "min_count":3,
                              "c_id": 1
                         }
         */
        JSONObject onlineHistoryProfile = paramJsonObject.getJSONObject("online_history_profile");
        String startTimeStr = onlineHistoryProfile.getString("start_time");
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        Date startTimeDate = sdf.parse(startTimeStr);
        // 查询条件的时间范围起始点
        long queryStartTime = startTimeDate.getTime();
        // 查询条件的时间范围截止点(此刻)
        long queryEndTime = System.currentTimeMillis();
        // 查询条件的目标事件
        String queryEventId = onlineHistoryProfile.getString("event_id");
        // 本条件的标识id
        Integer cId = onlineHistoryProfile.getInteger("c_id");

        Connection connection = DriverManager.getConnection("jdbc:mysql://doitedu:9030/test", "root", "root");
        String querySql =
                " SELECT                            "+
                "     user_id,                      "+
                "     count(1) as cnt               "+
                " FROM user_events                  "+
                " WHERE event_id = ?                "+
                " AND event_time >= ?               "+
                " AND event_time < ?                "+
                " GROUP BY user_id                  ";

        PreparedStatement stmt = connection.prepareStatement(querySql);
        stmt.setString(1,queryEventId);
        stmt.setLong(2,queryStartTime);
        stmt.setLong(3,queryEndTime);


        // 创建redis连接
        Jedis jedis = new Jedis("doitedu", 6379);

        // 执行查询
        ResultSet rs = stmt.executeQuery();
        while(rs.next()){
            long userId = rs.getLong("user_id");
            long cnt = rs.getLong("cnt");

            // 向redis中写入该用户的(前提是该用户属于目标人群）
            if(targetUsersBitmap.contains(userId)) {
                jedis.hset(ruleId + ":" + cId, userId + "", cnt + "");
            }
        }


        // 然后将规则的参数json，以及查询到的目标人群bitmap，写入规则系统的元数据表
        PreparedStatement pst = conn.prepareStatement("insert into rule_meta values (?,?,?,?,?,?,?,?,?)");
        pst.setString(1, ruleId);
        pst.setString(2, paramJson);

        // 放入规则所属的模型的运算机源代码
        pst.setString(3,calculatorSourceCode);

        pst.setInt(4, 1);
        pst.setBytes(5, targetUsersBitmapBytes);  // 预圈选目标人群

        pst.setLong(6,queryEndTime);  // 历史值查询范围截止时间点

        pst.setString(7, "深似海的男人");
        Timestamp timestamp = new Timestamp(System.currentTimeMillis());
        pst.setTimestamp(8, timestamp);
        pst.setTimestamp(9, timestamp);

        boolean execute = pst.execute();
        System.out.println(execute);

        return execute ? "新增规则失败" : "新增规则成功";
    }
}
